/*=====================================================================*
 *                   Copyright (C) 2011 Paul Mineiro                   *
 * All rights reserved.                                                *
 *                                                                     *
 * Redistribution and use in source and binary forms, with             *
 * or without modification, are permitted provided that the            *
 * following conditions are met:                                       *
 *                                                                     *
 *     * Redistributions of source code must retain the                *
 *     above copyright notice, this list of conditions and             *
 *     the following disclaimer.                                       *
 *                                                                     *
 *     * Redistributions in binary form must reproduce the             *
 *     above copyright notice, this list of conditions and             *
 *     the following disclaimer in the documentation and/or            *
 *     other materials provided with the distribution.                 *
 *                                                                     *
 *     * Neither the name of Paul Mineiro nor the names                *
 *     of other contributors may be used to endorse or promote         *
 *     products derived from this software without specific            *
 *     prior written permission.                                       *
 *                                                                     *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND              *
 * CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,         *
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES               *
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE             *
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER               *
 * OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,                 *
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES            *
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE           *
 * GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR                *
 * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF          *
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT           *
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY              *
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE             *
 * POSSIBILITY OF SUCH DAMAGE.                                         *
 *                                                                     *
 * Contact: Paul Mineiro <paul@mineiro.com>                            *
 *=====================================================================*/

#ifndef __SFT_PARSE_WORKER_HH__
#define __SFT_PARSE_WORKER_HH__

#include <boost/interprocess/detail/atomic.hpp>
#include <boost/optional.hpp>
#include <cstdlib>
#include "example.hh"
#include "parse.hh"
#include "threadqueue.hh"

namespace flassol
{
  class SftParseWorker
    {
      private:
        struct BadParse
          {
          };

        static unsigned int
        parse_maybe_uint (const char* s)
        {
          unsigned int rv = 0;

          if (s)
            {
              char* endptr;

              rv = strtoul (s, &endptr, 0);

              if (*endptr != '\0')
                {
                  throw BadParse ();
                }
            }

          return rv;
        }

        static SparseVector
        parse_sparse_vector (const char*  s,
                             unsigned int next_index,
                             unsigned int k)
        {
          SparseVector rv;

          // value(:index)?,value(:index)?...
          
          while (*s != '\0')
            {
              char* endptr;
              float value = strtof (s, &endptr);
              unsigned int index = next_index;

              if (*endptr == ':')
                {
                  index = strtoul (endptr + 1, &endptr, 0);
                }

              if (*endptr == ',')
                {
                  ++endptr;
                }
              else if (*endptr != '\0')
                {
                  throw BadParse ();
                }

              if (index >= k)
                {
                  throw BadParse ();
                }

              s = endptr;
              rv[index] = -value;
              next_index = index + 1;
            }

          return rv;
        }

        static SparseVector
        delta_sparse_vector (unsigned int index,
                             unsigned int k)
        {
          SparseVector rv;

          if (index >= k)
            {
              throw BadParse ();
            }

          rv[index] = -1;

          return rv;
        }

        static unsigned int
        log2ceil (unsigned int x)
          {
            unsigned int rv = 1;

            for (x >>= 1; x > 0; x >>= 1) { ++rv; }

            return rv;
          }


        static boost::optional<CSMCExample>
        from_general (const boost::optional<GeneralExample>& e,
                      unsigned int                           num_classes,
                      unsigned int                           num_consumers,
                      bool                                   adaptive)
          {
            using boost::interprocess::detail::atomic_write32;

            if (e)
              {
                try
                  {
                    CSMCExample r;

                    if (e->importance)
                      {
                        r.cost = 
                          parse_sparse_vector (e->importance,
                                               parse_maybe_uint (e->label),
                                               num_classes);
                      }
                    else if (e->label)
                      {
                        r.cost = 
                          delta_sparse_vector (parse_maybe_uint (e->label),
                                               num_classes);
                      }

                    if (r.cost)
                      {
                        if (num_consumers > 1)
                          {
                            if (adaptive)
                              {
                                r.norm_partials = 
                                  boost::shared_ptr<FloatMatrix>
                                    (new FloatMatrix);

                                r.norm_partials->resize (num_consumers);

                                for (unsigned int i = 0; 
                                     i < num_consumers; 
                                     ++i)
                                  {
                                    (*r.norm_partials)[i].resize (num_classes - 1);
                                  }

                                r.adaptivecount = 
                                  new boost::uint32_t[log2ceil (num_classes)];

                                for (unsigned int i = 0; 
                                     i < log2ceil (num_classes);
                                     ++i)
                                  {
                                    atomic_write32 (r.adaptivecount + i,
                                                    num_consumers);
                                  }
                              }

                            r.updatecount = 
                              new boost::uint32_t[log2ceil (num_classes)];

                            for (unsigned int i = 0; 
                                 i < log2ceil (num_classes);
                                 ++i)
                              {
                                atomic_write32 (r.updatecount + i,
                                                num_consumers);
                              }
                            

                            r.estimationcount = 
                              new boost::uint32_t[log2ceil (num_classes)];

                            for (unsigned int i = 0; 
                                 i < log2ceil (num_classes);
                                 ++i)
                              {
                                atomic_write32 (r.estimationcount + i,
                                                num_consumers);
                              }
                            }
                        }

                    r.tag = e->tag;
                    r.f = e->f;

                    r.initials = 
                      boost::shared_ptr<EstimateResultMatrix>
                        (new EstimateResultMatrix);

                    r.initials->resize (num_consumers);

                    for (unsigned int i = 0; i < num_consumers; ++i)
                      {
                        (*r.initials)[i].resize (num_classes - 1);
                      }

                    r.partials = 
                      boost::shared_ptr<EstimateResultMatrix>
                        (new EstimateResultMatrix);

                    r.partials->resize (num_consumers);

                    for (unsigned int i = 0; i < num_consumers; ++i)
                      {
                        (*r.partials)[i].resize (num_classes - 1);
                      }

                    if (num_consumers > 1)
                      {
                        r.usecount = 
                          new boost::uint32_t[log2ceil (num_classes)];

                        for (unsigned int i = 0; 
                             i < log2ceil (num_classes);
                             ++i)
                          {
                            atomic_write32 (r.usecount + i, 
                                            num_consumers);
                          }
                      }

                    r.buf = e->buf;

                    return r;
                  }
                catch (BadParse&)
                  {
                  }
              }

            return boost::none;
          }

        StandardParse parse;
        std::istream& in;
        Tee<CSMCQueue>& out;
        bool adaptive;
        unsigned int num_classes;

      public:
        SftParseWorker (StandardParse::hash_option_value ho,
                        std::istream&                    _in,
                        Tee<CSMCQueue>&                  _out,
                        bool                             _adaptive,
                        unsigned int                     _num_classes)
          : parse (ho),
            in (_in),
            out (_out),
            adaptive (_adaptive),
            num_classes (_num_classes)
          {
          }

        void operator() ()
          {
            std::vector<boost::optional<CSMCExample> >* examples = 0;

            out.add_writer ();

            while (in.good ())
              {
                boost::optional<GeneralExample> example = parse (in);
                boost::optional<CSMCExample> r = from_general (example,
                                                               num_classes,
                                                               out.num_consumers (),
                                                               adaptive);

                if (r || ! in.eof ())
                  {
                    if (! examples)
                      {
                        examples = 
                          new std::vector<boost::optional<CSMCExample> >;
                        examples->reserve (31);
                      }

                    examples->push_back (r);

                    if (examples->size () > 30)
                      {
                        out.push (examples);
                        examples = 0;
                      }
                  }
              }

            if (examples)
              {
                out.push (examples);
              }

            out.del_writer ();
          }
    };
}

#endif // __SFT_PARSE_WORKER_HH__
